/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.route;

import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;
import com.pingan.haofang.searchcloud.api.route.IndexAliasRegister;
import com.pingan.haofang.searchcloud.api.route.IndexRoute;
import org.apache.commons.collections.CollectionUtils;
import org.apache.solr.common.cloud.SolrZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.Collection;

/**
 * Collection注册服务zk实现
 *
 * @author huhailiang937
 * @date 2017年4月7日下午4:46:15
 * @since 1.0.0
 */
public class Processor4ZKIndexRegister {

    protected static Logger LOG = LoggerFactory.getLogger(Processor4ZKIndexRegister.class);

    private SolrZkClient zkClient;

    private String metaZkHost;

    private Collection<IndexAliasRegister> indexAliasRegisters;

    public Processor4ZKIndexRegister(String metaZkHost, Collection<IndexAliasRegister> indexAliasRegisters) {
        this.metaZkHost = metaZkHost;
        this.zkClient = new SolrZkClient(metaZkHost,30000);
        this.indexAliasRegisters = indexAliasRegisters;
    }

    public boolean register(IndexRoute indexRoute) {
        Assert.notNull(indexRoute, "collectionRoute can not null");
        Assert.notNull(indexRoute.getCollectionRouteKey(), "CollectionRouteKey can not null");
        Assert.notNull(indexRoute.getCollectionName(), "CollectionName can not null");

        String collectionRoutePath = MetaCacheSupport.getZKRoutePath(indexRoute.getCollectionRouteKey());

        
        if(!zkClient.isConnected()){
            zkClient.close();
            zkClient = new SolrZkClient(metaZkHost,30000);
        }
        
        boolean initSuccess = initZKPath(zkClient, collectionRoutePath);
        if (!initSuccess) {
            return false;
        }

        try {
            String json = new String(MetaCacheSupport.encode(indexRoute));
            zkClient.setData(collectionRoutePath, json.getBytes(), true);
            LOG.info(String.format("ZKCollectionRouteRegister register CollectionRoute : %s", json));

            if (CollectionUtils.isNotEmpty(indexAliasRegisters)) {
                for (IndexAliasRegister aliasRegister : indexAliasRegisters) {
                    aliasRegister.register(indexRoute);
                }
            }

            return true;
        } catch (Exception e) {
            LOG.error(String.format("ZKCollectionRouteRegister register fail! setData(%s) has error:",
                    collectionRoutePath), e);
        }
        return false;
    }


    private boolean initZKPath(SolrZkClient zkClient, String collectionRoutePath) {
        try {
            boolean isExists = zkClient.exists(collectionRoutePath, false);
            if (!isExists) {
                zkClient.makePath(collectionRoutePath, true, false);
            }
            return true;
        } catch (Exception e) {
            LOG.error(String.format("zkClient makePath(%s) has error:", collectionRoutePath), e);
        }
        return false;
    }

}
